package com.shiguiwu.springboot3.kafka.listener;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * @description: 监听消费消息
 * @author: stone
 * @date: Created by 2023/11/13 16:38
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.kafka.listener
 */
@Component
@Slf4j
public class KafkaConsumerListener {

    //监听消费
    @KafkaListener(topics = {"shigwTopic"})
    public void multiPartitionsOn(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        try {
            log.warn("第一个客户端topic = {},record.partition={},value={} ", record.topic(), record.partition(), record.value());
        } finally {
            Optional.ofNullable(ack).ifPresent(Acknowledgment::acknowledge);
        }

    }

    //监听消费
    @KafkaListener(topics = {"test520"})
    public void onNormalMessage(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        try {
            log.warn("topic = {},record.partition={},value={} ", record.topic(), record.partition(), record.value());
        } finally {
            Optional.ofNullable(ack).ifPresent(Acknowledgment::acknowledge);
        }

    }

    //批量消费
    @KafkaListener(id = "aaa",topics = {"test002"}, groupId = "520_group")
    public void onBatchMessage(List<ConsumerRecord<String, Object>> records) {
        log.error(">>> 批量消费一次，records.size()= {}", records.size());
        for (ConsumerRecord<String, Object> record : records) {
            log.error("嘻嘻 ====> {}",record.value());
        }
    }

    /**
     * @param record 消息
     *  手动ack
     */
    //@KafkaListener(
    //        topics = {"#{'${spring.kafka.bindings.analysis.destination}'}"}
    //        , containerFactory = "kafkaListenerContainerFactory"
    //        , autoStartup = "#{'${spring.kafka.bindings.analysis.autoStartup}'}"
    //        , groupId = "#{'${spring.kafka.bindings.analysis.group}'}"
    //        , concurrency = "#{'${spring.kafka.bindings.analysis.concurrency}'}")
    public void receiveMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String message = record.value();
        try {
            JSONObject jsonObject = JSONUtil.parseObj(message);
            log.info("告警开始处理：{}", message);
            //iAnalysisRiskAppService.stream(jsonObject);
        } catch (Exception e) {
            log.error("告警处理异常：{}", message, e);
        } finally {
            if (ack != null) {
                ack.acknowledge();
            }
        }

    }

    //消息转发 从sb_topic转发到sb_topic2
    //@KafkaListener(topics = {"test520"},groupId = "521_group")
    //@SendTo("test002")
    //public String onMessage7(ConsumerRecord<?, ?> record) {
    //    return record.value()+"-forward message";
    //}
    //
    //@KafkaListener(topics = {"test5201"})
    //public void onMessage8(ConsumerRecord<?, ?> record) {
    //    System.out.println("test520：" + record.value());
    //}


}
